package com.rs.lisenter;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.rs.config.IdWorker;
import com.rs.config.PadConfig;
import com.rs.data.model.PadPushData;
import com.rs.service.IPadExchangeHelpService;
import org.apache.kafka.clients.consumer.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * Kafka消费监听
 */
@Component
public class KafkaConsumer {

    private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @Resource
    private IPadExchangeHelpService padExchangeHelpService;

    @KafkaListener(topics = {"#{'${spring.kafka.call-center.topic}'.split(',')}"}, groupId = "G1", containerFactory = "manualKafkaListenerContainerFactory")
    public void manualCommit(@Payload String message,
                             @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                             @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                             Consumer consumer,
                             Acknowledgment ack) {
        logger.info("manualCommit手动提交偏移量, topic={}, partition={}, msg={}", topic, partition, message);

        IdWorker worker = new IdWorker(1, 1);

        JSONObject jsonObject = JSONObject.parseObject(message);
        for (String key : jsonObject.keySet()) {
            String obj = jsonObject.getString(key);
            if (StrUtil.isEmpty(obj) || !"tableSchema".equals(key)) {
                continue;
            }
            Optional.ofNullable(jsonObject.getJSONObject("tableSchema")).ifPresent(tableSchema -> {
                if (tableSchema.containsKey("hasFile") && "true".equals(tableSchema.getString("hasFile"))) {
                    // 附件路径字段
                    String path = jsonObject.getJSONObject("tableSchema").getString("path");
                    // 附件路径前缀
                    String prefix = jsonObject.getJSONObject("tableSchema").getString("prefix");
                    PadPushData pushData = new PadPushData(
                            String.valueOf(worker.nextId()),
                            "",
                            PadConfig.ACTION_ATTACHMENT,
                            "",
                            prefix + jsonObject.getString(path));
                    // 向推送webservice推送数据
                    padExchangeHelpService.addData(pushData);
                }
            });
        }

        PadPushData pushData = new PadPushData(
                String.valueOf(worker.nextId()),
                "",
                PadConfig.ACTION_ADD,
                "",
                message);
        // 向推送webservice推送数据
        padExchangeHelpService.addData(pushData);

        // 同步提交
        consumer.commitSync();
    }
}
